package com.dmp.storm.topology;

import org.dom4j.Document;
import org.dom4j.Element;
import org.dom4j.io.SAXReader;

import java.io.FileInputStream;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class XmlTopologyCtx extends BaseTopologyCtx{

	public XmlTopologyCtx(String... paths) {
		super();
		if (paths.length == 0) {
			paths = new String[] { "storm.xml" };
		}
		try {
			for (String xmlPath : paths) {
				this.createWorkFlow(SS.LOCAL, xmlPath);
			}
		} catch (Exception e) {
			e.printStackTrace();
			System.exit(-1);
		}
	}
	public XmlTopologyCtx(BaseTopologyCtx.SS ss, String... paths) {
		super();
		if (paths.length == 0) {
			paths = new String[] { "storm.xml" };
		}
		try {
			for (String xmlPath : paths) {
				this.createWorkFlow(ss, xmlPath);
			}

		} catch (Exception e) {
			e.printStackTrace();
			System.exit(-1);
		}
	}
	private Map<String,String> getGroupType(Element bolt){
		Map<String,String> groupMap = new HashMap<String, String>();
		String group = bolt.attributeValue("shuffle");
		if (group != null) {
			groupMap.put("type", "shuffle");
			groupMap.put("field", group);
			return groupMap;
		}
		group = bolt.attributeValue("custom");
		if (group != null) {
			groupMap.put("type", "custom");
			groupMap.put("field", group);
			return groupMap;
		}
		group = bolt.attributeValue("fields");
		if (group != null) {
			groupMap.put("type", "fields");
			groupMap.put("field", group);
			return groupMap;
		}
		group = bolt.attributeValue("direct");
		if (group != null) {
			groupMap.put("type", "direct");
			groupMap.put("field", group);
			return groupMap;
		}
		return null;
	}
	@SuppressWarnings("unchecked")
	private Map<String,String> getProps(Element ele){
		Map<String,String> propsMap = new HashMap<String, String>();
		List<Element> props = ele.elements("property");
		for (Element prop : props) {
			String attrKey = prop.attributeValue("name");
			String attrValue = prop.attributeValue("value");
			propsMap.put(attrKey, attrValue);
		}
		return propsMap;
	}
	@SuppressWarnings("unchecked")
	@Override
	protected void createWorkFlow(BaseTopologyCtx.SS ss, String path) {
		SAXReader reader = new SAXReader();
		Document doc = null;
		try {
			if(BaseTopologyCtx.SS.LOCAL.equals(ss)){
				doc = reader.read(XmlTopologyCtx.class.getClassLoader()
						.getResourceAsStream(path));
			}else{
				doc = reader.read(new FileInputStream(path));
			}

		} catch (Exception e) {
			System.err.println("加载工作流出现异常");
			e.printStackTrace();
			System.exit(-1);
		}
		Element root = doc.getRootElement();
		List<Element> groupingList = root.elements("grouping");
		if (!groupingList.isEmpty()) {
			for (Element e : groupingList) {
				String groupname = e.attributeValue("name");
				String classPath = e.attributeValue("class");
				Map<String,String> props = this.getProps(e);
				this.createStreamGrouping(groupname, classPath, props);
			}
		}
		List<Element> spoutList = root.elements("spout");
		if (!spoutList.isEmpty()) {
			for (Element e : spoutList) {
				String spoutname = e.attributeValue("name");
				String classPath = e.attributeValue("class");
				String num = e.attributeValue("paralleNum");
				Map<String,String> props = this.getProps(e);
				this.createSpout(spoutname, classPath, props,num);
			}
		}
		List<Element> boltList = root.elements("bolt");
		if (!boltList.isEmpty()) {
			for (Element bolt: boltList) {
				String boltname = bolt.attributeValue("name");
				String classPath = bolt.attributeValue("class");
				String num = bolt.attributeValue("paralleNum");
				Map<String,String> props = this.getProps(bolt);
				Map<String,String> groupMap = this.getGroupType(bolt);
				this.createBolt(boltname, classPath, props,num,groupMap.get("type"),groupMap.get("field"));
			}
		}
		List<Element> importList = root.elements("import");
		if (!importList.isEmpty()) {
			for (Element e : boltList) {
				this.createWorkFlow(ss, e.attributeValue("src"));
			}
		}
	}
}